// This file is part of OpenTSDB.
// Copyright (C) 2013  The OpenTSDB Authors.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tree;

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.powermock.api.mockito.PowerMockito.mock;

import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import net.opentsdb.core.TSDB;
import net.opentsdb.storage.MockBase;
import net.opentsdb.utils.Config;
import net.opentsdb.utils.JSON;

import org.hbase.async.DeleteRequest;
import org.hbase.async.GetRequest;
import org.hbase.async.HBaseClient;
import org.hbase.async.KeyValue;
import org.hbase.async.PutRequest;
import org.hbase.async.Scanner;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;

@RunWith(PowerMockRunner.class)
@PowerMockIgnore({"javax.management.*", "javax.xml.*",
                  "ch.qos.*", "org.slf4j.*",
                  "com.sum.*", "org.xml.*"})
@PrepareForTest({ TSDB.class, HBaseClient.class, GetRequest.class,
  PutRequest.class, KeyValue.class, Scanner.class, DeleteRequest.class })
public final class TestBranchStage {
  private final static byte[] NAME_FAMILY = "name".getBytes(MockBase.ASCII());
  private final static byte[] TREE_TABLE = "tsdb-tsdbTree".getBytes(MockBase.ASCII());
  private final static byte[] UID_TABLE = "tsdb-uid".getBytes(MockBase.ASCII());
  private MockBase storage;
  private TsdbTree tsdbTree = TestTsdbTree.buildTestTree();
  final static private Method toStorageJson;
  static {
    try {
      toStorageJson = BranchStage.class.getDeclaredMethod("toStorageJson");
      toStorageJson.setAccessible(true);
    } catch (Exception e) {
      throw new RuntimeException("Failed in static initializer", e);
    }
  }
  
  final static private Method LeaftoStorageJson;
  static {
    try {
      LeaftoStorageJson = Leaf.class.getDeclaredMethod("toStorageJson");
      LeaftoStorageJson.setAccessible(true);
    } catch (Exception e) {
      throw new RuntimeException("Failed in static initializer", e);
    }
  }
  
  @Test
  public void copyConstructor() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);
    final BranchStage copy = new BranchStage(branchStage);
    assertEquals(1, copy.getTreeId());
    assertEquals("ROOT", copy.getDisplayName());
    assertNotNull(copy.getBranchStages());
    assertTrue(copy.getBranchStages() != branchStage.getBranchStages());
    assertNotNull(copy.getLeaves());
    assertTrue(copy.getLeaves() != branchStage.getLeaves());
    assertNotNull(copy.getPath());
    assertTrue(copy.getPath() != branchStage.getPath());
  }
  
  @Test
  public void testHashCode() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);
    assertEquals(2521314, branchStage.hashCode());
  }
  
  @Test
  public void testEquals() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    final BranchStage branchStage2 = buildTestBranch(tsdbTree);;
    assertTrue(branchStage.equals(branchStage2));
  }
  
  @Test
  public void equalsSameAddress() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertTrue(branchStage.equals(branchStage));
  }
  
  @Test
  public void equalsNull() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertFalse(branchStage.equals(null));
  }
  
  @Test
  public void equalsWrongClass() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertFalse(branchStage.equals(new Object()));
  }
  
  @Test
  public void compareTo() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    final BranchStage branchStage2 = buildTestBranch(tsdbTree);;
    assertEquals(0, branchStage.compareTo(branchStage2));
  }
  
  @Test
  public void compareToLess() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    final BranchStage branchStage2 = buildTestBranch(tsdbTree);;
    branchStage2.setDisplayName("Ardvark");
    assertTrue(branchStage.compareTo(branchStage2) > 0);
  }
  
  @Test
  public void compareToGreater() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    final BranchStage branchStage2 = buildTestBranch(tsdbTree);;
    branchStage2.setDisplayName("Zelda");
    assertTrue(branchStage.compareTo(branchStage2) < 0);
  }
  
  @Test
  public void getBranchIdRoot() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertEquals("0001", branchStage.getBranchId());
  }
  
  @Test
  public void getBranchIdChild() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertEquals("0001D119F20E", branchStage.getBranchStages().first().getBranchId());
  }
  
  @Test
  public void addChild() throws Exception {
    final BranchStage branchStage = buildTestBranch(tsdbTree);
    final BranchStage child = new BranchStage(tsdbTree.getTreeId());
    assertTrue(branchStage.addChild(child));
    assertEquals(3, branchStage.getBranchStages().size());
    assertEquals(2, branchStage.getLeaves().size());
  }
  
  @Test
  public void addChildNoLocalBranches() throws Exception {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    final BranchStage child = new BranchStage(tsdbTree.getTreeId());
    Field branches = BranchStage.class.getDeclaredField("branches");
    branches.setAccessible(true);
    branches.set(branchStage, null);
    branches.setAccessible(false);
    assertTrue(branchStage.addChild(child));
    assertEquals(1, branchStage.getBranchStages().size());
    assertEquals(2, branchStage.getLeaves().size());
  }
  
  @Test
  public void addChildNoChanges() throws Exception {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    final BranchStage child = new BranchStage(tsdbTree.getTreeId());
    assertTrue(branchStage.addChild(child));
    assertFalse(branchStage.addChild(child));
    assertEquals(3, branchStage.getBranchStages().size());
    assertEquals(2, branchStage.getLeaves().size());
  }
  
  @Test
  public void addLeafExists() throws Exception {
    final TsdbTree tsdbTree = TestTsdbTree.buildTestTree();
    final BranchStage branchStage = buildTestBranch(tsdbTree);;

    Leaf leaf = new Leaf();
    leaf.setDisplayName("Alarms");
    leaf.setTsuid("ABCD");

    assertFalse(branchStage.addLeaf(leaf, tsdbTree));
    assertEquals(2, branchStage.getBranchStages().size());
    assertEquals(2, branchStage.getLeaves().size());
    assertNull(tsdbTree.getCollisions());
  }
  
  @Test
  public void addLeafCollision() throws Exception {
    final TsdbTree tsdbTree = TestTsdbTree.buildTestTree();
    final BranchStage branchStage = buildTestBranch(tsdbTree);;

    Leaf leaf = new Leaf();
    leaf.setDisplayName("Alarms");
    leaf.setTsuid("0001");

    assertFalse(branchStage.addLeaf(leaf, tsdbTree));
    assertEquals(2, branchStage.getBranchStages().size());
    assertEquals(2, branchStage.getLeaves().size());
    assertEquals(1, tsdbTree.getCollisions().size());
  }

  @Test (expected = IllegalArgumentException.class)
  public void addChildNull() throws Exception {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    branchStage.addChild(null);
  }

  @Test
  public void addLeaf() throws Exception {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    
    Leaf leaf = new Leaf();
    leaf.setDisplayName("Application Servers");
    leaf.setTsuid("0004");
    
    assertTrue(branchStage.addLeaf(leaf, null));
  }

  @Test (expected = IllegalArgumentException.class)
  public void addLeafNull() throws Exception {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    branchStage.addLeaf(null, null);
  }
  
  @Test
  public void compileBranchId() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertArrayEquals(new byte[] { 0, 1 }, branchStage.compileBranchId());
  }
  
  @Test
  public void compileBranchIdChild() {
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    assertArrayEquals(new byte[] { 0, 1 , (byte) 0xD1, 0x19, (byte) 0xF2, 0x0E }, 
        branchStage.getBranchStages().first().compileBranchId());
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void compileBranchIdEmptyDisplayName() {
    final BranchStage branchStage = new BranchStage(1);
    branchStage.compileBranchId();
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void compileBranchIdInvalidId() {
    final BranchStage branchStage = new BranchStage(0);
    branchStage.compileBranchId();
  }

  @Test
  public void fetchBranch() throws Exception {
    setupStorage();
    
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 1 }, 
        NAME_FAMILY,
        "metrics".getBytes(MockBase.ASCII()),
        "sys.cpu.0".getBytes(MockBase.ASCII()));
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 1 }, 
        NAME_FAMILY,
        "tagk".getBytes(MockBase.ASCII()),
        "host".getBytes(MockBase.ASCII()));
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 1 }, 
        NAME_FAMILY,
        "tagv".getBytes(MockBase.ASCII()),
        "web01".getBytes(MockBase.ASCII()));
    
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 2 }, 
        NAME_FAMILY,
        "metrics".getBytes(MockBase.ASCII()),
        "sys.cpu.1".getBytes(MockBase.ASCII()));
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 2 }, 
        NAME_FAMILY,
        "tagk".getBytes(MockBase.ASCII()),
        "owner".getBytes(MockBase.ASCII()));
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 2 }, 
        NAME_FAMILY,
        "tagv".getBytes(MockBase.ASCII()),
        "ops".getBytes(MockBase.ASCII()));
    
    final BranchStage branchStage = BranchStage.fetchBranch(storage.getTSDB(),
        BranchStage.stringToId("00010001BECD000181A8"), true).joinUninterruptibly();
    assertNotNull(branchStage);
    assertEquals(1, branchStage.getTreeId());
    assertEquals("cpu", branchStage.getDisplayName());
    assertEquals("00010001BECD000181A8", branchStage.getBranchId());
    assertEquals(1, branchStage.getBranchStages().size());
    assertEquals(2, branchStage.getLeaves().size());
  }
  
  @Test
  public void fetchBranchNSU() throws Exception {
    setupStorage();
    
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 1 }, 
        NAME_FAMILY,
        "metrics".getBytes(MockBase.ASCII()),
        "sys.cpu.0".getBytes(MockBase.ASCII()));
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 1 }, 
        NAME_FAMILY,
        "tagk".getBytes(MockBase.ASCII()),
        "host".getBytes(MockBase.ASCII()));
    storage.addColumn(UID_TABLE, new byte[] { 0, 0, 1 }, 
        NAME_FAMILY,
        "tagv".getBytes(MockBase.ASCII()),
        "web01".getBytes(MockBase.ASCII()));
    
    final BranchStage branchStage = BranchStage.fetchBranch(storage.getTSDB(),
        BranchStage.stringToId("00010001BECD000181A8"), true).joinUninterruptibly();
    assertNotNull(branchStage);
    assertEquals(1, branchStage.getTreeId());
    assertEquals("cpu", branchStage.getDisplayName());
    assertEquals("00010001BECD000181A8", branchStage.getBranchId());
    assertEquals(1, branchStage.getBranchStages().size());
    assertEquals(1, branchStage.getLeaves().size());
  }
  
  @Test
  public void fetchBranchNotFound() throws Exception {
    setupStorage();
    final BranchStage branchStage = BranchStage.fetchBranch(storage.getTSDB(),
        BranchStage.stringToId("00010001BECD000181A0"), false).joinUninterruptibly();
    assertNull(branchStage);
  }
  
  @Test
  public void fetchBranchOnly() throws Exception {
    setupStorage();
    final BranchStage branchStage = BranchStage.fetchBranchOnly(storage.getTSDB(),
        BranchStage.stringToId("00010001BECD000181A8")).joinUninterruptibly();
    assertNotNull(branchStage);
    assertEquals("cpu", branchStage.getDisplayName());
    assertNull(branchStage.getLeaves());
    assertNull(branchStage.getBranchStages());
  }

  @Test
  public void fetchBranchOnlyNotFound() throws Exception {
    setupStorage();
    final BranchStage branchStage = BranchStage.fetchBranchOnly(storage.getTSDB(),
        BranchStage.stringToId("00010001BECD000181A0")).joinUninterruptibly();
    assertNull(branchStage);
  }
  
  @Test
  public void storeBranch() throws Exception {
    setupStorage();
    final BranchStage branchStage = buildTestBranch(tsdbTree);
    branchStage.storeBranch(storage.getTSDB(), tsdbTree, true);
    assertEquals(3, storage.numRows(TREE_TABLE));
    assertEquals(3, storage.numColumns(TREE_TABLE, new byte[] { 0, 1 }));
    final BranchStage parsed = JSON.parseToObject(storage.getColumn(TREE_TABLE,
        new byte[] { 0, 1 }, TsdbTree.TREE_FAMILY(),
        "branchStage".getBytes(MockBase.ASCII())), BranchStage.class);
    parsed.setTreeId(1);
    assertEquals("ROOT", parsed.getDisplayName());
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void storeBranchMissingTreeID() throws Exception {
    setupStorage();
    final BranchStage branchStage = new BranchStage();
    branchStage.storeBranch(storage.getTSDB(), tsdbTree, false);
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void storeBranchTreeID0() throws Exception {
    setupStorage();
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    branchStage.setTreeId(0);
    branchStage.storeBranch(storage.getTSDB(), tsdbTree, false);
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void storeBranchTreeID65536() throws Exception {
    setupStorage();
    final BranchStage branchStage = buildTestBranch(tsdbTree);;
    branchStage.setTreeId(65536);
    branchStage.storeBranch(storage.getTSDB(), tsdbTree, false);
  }

  @Test
  public void storeBranchExistingLeaf() throws Exception {
    setupStorage();
    final BranchStage branchStage = buildTestBranch(tsdbTree);
    Leaf leaf = new Leaf("Alarms", "ABCD");
    byte[] qualifier = leaf.columnQualifier();
    storage.addColumn(branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        qualifier, (byte[])LeaftoStorageJson.invoke(leaf));
    
    branchStage.storeBranch(storage.getTSDB(), tsdbTree, true);
    assertEquals(3, storage.numRows(TREE_TABLE));
    assertEquals(3, storage.numColumns(TREE_TABLE, new byte[] { 0, 1 }));
    assertNull(tsdbTree.getCollisions());
    final BranchStage parsed = JSON.parseToObject(storage.getColumn(TREE_TABLE,
        new byte[] { 0, 1 }, TsdbTree.TREE_FAMILY(),
        "branchStage".getBytes(MockBase.ASCII())), BranchStage.class);
    parsed.setTreeId(1);
    assertEquals("ROOT", parsed.getDisplayName());
  }
  
  @Test
  public void storeBranchCollision() throws Exception {
    setupStorage();
    final BranchStage branchStage = buildTestBranch(tsdbTree);
    Leaf leaf = new Leaf("Alarms", "0101");
    byte[] qualifier = leaf.columnQualifier();
    storage.addColumn(TREE_TABLE, branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        qualifier, (byte[])LeaftoStorageJson.invoke(leaf));
    
    branchStage.storeBranch(storage.getTSDB(), tsdbTree, true);
    assertEquals(3, storage.numRows(TREE_TABLE));
    assertEquals(3, storage.numColumns(TREE_TABLE, new byte[] { 0, 1 }));
    assertEquals(1, tsdbTree.getCollisions().size());
    final BranchStage parsed = JSON.parseToObject(storage.getColumn(TREE_TABLE,
        new byte[] { 0, 1 }, TsdbTree.TREE_FAMILY(),
        "branchStage".getBytes(MockBase.ASCII())), BranchStage.class);
    parsed.setTreeId(1);
    assertEquals("ROOT", parsed.getDisplayName());
  }
  
  @Test
  public void idToString() throws Exception {
    assertEquals("0EA8", BranchStage.idToString(new byte[] { 0x0E, (byte) 0xA8 }));
  }
  
  @Test
  public void idToStringZeroes() throws Exception {
    assertEquals("0000", BranchStage.idToString(new byte[] { 0, 0 }));
  }
  
  @Test (expected = NullPointerException.class)
  public void idToStringNull() throws Exception {
    BranchStage.idToString(null);
  }
  
  @Test
  public void stringToId() throws Exception {
    assertArrayEquals(new byte[] { 0x0E, (byte) 0xA8 }, 
        BranchStage.stringToId("0EA8"));
  }
  
  @Test
  public void stringToIdZeros() throws Exception {
    assertArrayEquals(new byte[] { 0, 0 }, BranchStage.stringToId("0000"));
  }
  
  @Test
  public void stringToIdZerosPadding() throws Exception {
    assertArrayEquals(new byte[] { 0, 0, 0 }, BranchStage.stringToId("00000"));
  }
  
  @Test
  public void stringToIdCase() throws Exception {
    assertArrayEquals(new byte[] { 0x0E, (byte) 0xA8 }, 
        BranchStage.stringToId("0ea8"));
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void stringToIdNull() throws Exception {
    BranchStage.stringToId(null);
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void stringToIdEmpty() throws Exception {
    BranchStage.stringToId("");
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void stringToIdTooShort() throws Exception {
    BranchStage.stringToId("01");
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void stringToIdNotHex() throws Exception {
    BranchStage.stringToId("HelloWorld!");
  }
  
  @Test
  public void BRANCH_QUALIFIER() throws Exception {
    assertArrayEquals("branch".getBytes(MockBase.ASCII()), 
        BranchStage.BRANCH_QUALIFIER());
  }
  
  @Test
  public void prependParentPath() throws Exception {
    BranchStage branchStage = new BranchStage(1);
    branchStage.setDisplayName("cpu");
    final TreeMap<Integer, String> path = new TreeMap<Integer, String>();
    path.put(0, "ROOT");
    path.put(1, "sys");
    branchStage.prependParentPath(path);
       
    final Map<Integer, String> compiled_path = branchStage.getPath();
    assertNotNull(compiled_path);
    assertEquals(3, compiled_path.size());
  }
  
  @Test
  public void prependParentPathEmpty() throws Exception {
    BranchStage branchStage = new BranchStage(1);
    branchStage.setDisplayName("cpu");
    final TreeMap<Integer, String> path = new TreeMap<Integer, String>();
    branchStage.prependParentPath(path);
       
    final Map<Integer, String> compiled_path = branchStage.getPath();
    assertNotNull(compiled_path);
    assertEquals(1, compiled_path.size());
  }
  
  @Test (expected = IllegalArgumentException.class)
  public void prependParentPathNull() throws Exception {
    new BranchStage().prependParentPath(null);
  }
  
  /**
   * Helper to build a default branch for testing
   * @return A branch with some child branches and leaves
   */
  public static BranchStage buildTestBranch(final TsdbTree tsdbTree) {
    final TreeMap<Integer, String> root_path = new TreeMap<Integer, String>();
    final BranchStage root = new BranchStage(tsdbTree.getTreeId());
    root.setDisplayName("ROOT");
    root_path.put(0, "ROOT");
    root.prependParentPath(root_path);
    
    BranchStage child = new BranchStage(1);
    child.prependParentPath(root_path);
    child.setDisplayName("System");
    root.addChild(child);
    
    child = new BranchStage(tsdbTree.getTreeId());
    child.prependParentPath(root_path);
    child.setDisplayName("Network");
    root.addChild(child);

    Leaf leaf = new Leaf("Alarms", "ABCD");
    root.addLeaf(leaf, tsdbTree);
    
    leaf = new Leaf("Employees in Office", "EF00");
    root.addLeaf(leaf, tsdbTree);

    return root;
  }
  
  /**
   * Mocks classes for testing the storage calls
   */
  private void setupStorage() throws Exception {
    final HBaseClient client = mock(HBaseClient.class);
    final Config config = new Config(false);
    storage = new MockBase(new TSDB(client, config), 
        client, true, true, true, true);
    final List<byte[]> families = new ArrayList<byte[]>();
    families.add(TsdbTree.TREE_FAMILY());
    storage.addTable(TREE_TABLE, families);
    
    BranchStage branchStage = new BranchStage(1);
    TreeMap<Integer, String> path = new TreeMap<Integer, String>();
    path.put(0, "ROOT");
    path.put(1, "sys");
    path.put(2, "cpu");
    branchStage.prependParentPath(path);
    branchStage.setDisplayName("cpu");
    storage.addColumn(TREE_TABLE, branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        "branchStage".getBytes(MockBase.ASCII()),
        (byte[])toStorageJson.invoke(branchStage));
    
    Leaf leaf = new Leaf("user", "000001000001000001");
    byte[] qualifier = leaf.columnQualifier();
    storage.addColumn(TREE_TABLE, branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        qualifier, (byte[])LeaftoStorageJson.invoke(leaf));
    
    leaf = new Leaf("nice", "000002000002000002");
    qualifier = leaf.columnQualifier();
    storage.addColumn(TREE_TABLE, branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        qualifier, (byte[])LeaftoStorageJson.invoke(leaf));
    
    // child branchStage
    branchStage = new BranchStage(1);
    path.put(3, "mboard");
    branchStage.prependParentPath(path);
    branchStage.setDisplayName("mboard");
    storage.addColumn(TREE_TABLE, branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        "branchStage".getBytes(MockBase.ASCII()),
        (byte[])toStorageJson.invoke(branchStage));
    
    leaf = new Leaf("Asus", "000003000003000003");
    qualifier = leaf.columnQualifier();
    storage.addColumn(TREE_TABLE, branchStage.compileBranchId(), TsdbTree.TREE_FAMILY(),
        qualifier, (byte[])LeaftoStorageJson.invoke(leaf));
  }
}
